package com.bodsite.rpc.thrift.shard.stratrgy;

import java.util.Collection;
import java.util.LinkedHashSet;

import com.bodsite.common.logger.Logger;
import com.bodsite.common.logger.LoggerFactory;
import com.dangdang.ddframe.rdb.sharding.api.ShardingValue;
import com.dangdang.ddframe.rdb.sharding.api.strategy.database.SingleKeyDatabaseShardingAlgorithm;
import com.google.common.collect.Range;

/**
 * 数据源分两片，片0 表 0-9 片1 表10-19
 * @author bod
 * 2017年4月20日下午3:53:29
 */
public class BodsiteDatabaseShardingAlgorithm implements SingleKeyDatabaseShardingAlgorithm<Long>{
	private final static Logger logger = LoggerFactory.getLogger(BodsiteDatabaseShardingAlgorithm.class);
	/**
	    *  select * from t_order from t_order where order_id = 11 
	    *          └── SELECT *  FROM t_order_1 WHERE order_id = 11
	    *  select * from t_order from t_order where order_id = 44
	    *          └── SELECT *  FROM t_order_0 WHERE order_id = 44
	    */
	public String doEqualSharding(final Collection<String> tableNames, final ShardingValue<Long> shardingValue) {
		for (String each : tableNames) {
			if (each.endsWith(getModulo(shardingValue.getValue()))) {
				logger.debug(each+" "+shardingValue.getValue());
				return each;
			}
		}
		throw new IllegalArgumentException();
	}
	
	private String getModulo(Long value){
		return value % 20<10?"0":"1";
	}

	   /**
	    *  select * from t_order from t_order where order_id in (11,44)  
	    *          ├── SELECT *  FROM t_order_0 WHERE order_id IN (11,44) 
	    *          └── SELECT *  FROM t_order_1 WHERE order_id IN (11,44) 
	    *  select * from t_order from t_order where order_id in (11,13,15)  
	    *          └── SELECT *  FROM t_order_1 WHERE order_id IN (11,13,15)  
	    *  select * from t_order from t_order where order_id in (22,24,26)  
	    *          └──SELECT *  FROM t_order_0 WHERE order_id IN (22,24,26) 
	    */
	public Collection<String> doInSharding(final Collection<String> tableNames,
			final ShardingValue<Long> shardingValue) {
		Collection<String> result = new LinkedHashSet<>(tableNames.size());
		for (Long value : shardingValue.getValues()) {
			for (String tableName : tableNames) {
				if (tableName.endsWith(getModulo(value))) {
					result.add(tableName);
				}
			}
		}
		return result;
	}

	 /**
	    *  select * from t_order from t_order where order_id between 10 and 20 
	    *          ├── SELECT *  FROM t_order_0 WHERE order_id BETWEEN 10 AND 20 
	    *          └── SELECT *  FROM t_order_1 WHERE order_id BETWEEN 10 AND 20 
	    */
	public Collection<String> doBetweenSharding(final Collection<String> tableNames,
			final ShardingValue<Long> shardingValue) {
		Collection<String> result = new LinkedHashSet<>(tableNames.size());
		Range<Long> range = (Range<Long>) shardingValue.getValueRange();
		for (Long i = range.lowerEndpoint(); i <= range.upperEndpoint(); i++) {
			for (String each : tableNames) {
				if (each.endsWith(getModulo(i))) {
					result.add(each);
				}
			}
		}
		return result;
	}
}
